package jobs.slow_project_byte_summary;

import java.io.IOException;
import java.util.UUID;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


public class SlowProjectByteSummariserMap extends Mapper<LongWritable, Text, Text, LongWritable> {

	private UUID uniqueId = null;

	private Text project = new Text();
	private LongWritable bytes = new LongWritable();

	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

		if (uniqueId == null) {
			this.uniqueId = UUID.randomUUID();
		}


		try {
		
			if ((value.charAt(0) == '*') && (value.charAt(1) == ' ')) 
			{
		
				//System.out.println(uniqueId + ": Mapper processing line: " + value.toString());
		
				Long delayMilli = context.getConfiguration().getLong("delay_ms",0L);
				Integer delayNano = context.getConfiguration().getInt("delay_ns", 0);
				try {
					//System.out.println("Sleeping for " + delay);
					Thread.sleep(delayMilli,delayNano);
				}
				catch (InterruptedException ex) {
					ex.printStackTrace();
				}
				
				String line = value.toString().substring(2);
				String[] components = line.split(" ");
		
				project.set(components[0]);
				bytes.set(Long.valueOf(components[3]));
		
		
				context.write(project, bytes);
				
			}
			else {
				System.out.println(uniqueId + ": Mapper skipping incomplete line: " + value.toString());
			}
		
		
		}
		catch (Exception ex) {
			System.out.println(uniqueId + ": Mapper encountered exception. " + ex.getMessage());
		}
	}
}
	

